package com.bkhech.bigdata.tiktok_anchor.dataclean;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.Arrays;
import java.util.Objects;

/**
 * {
 *     "id":"158008900486",
 *     "uid":"120010010496",
 *     "nickname":"jack486",
 *     "gold":496,
 *     "watchnumpv":4860,
 *     "watchnumuv":972,
 *     "hots":1458,
 *     "nofollower":486,
 *     "looktime":9720,
 *     "smlook":2430,
 *     "follower":1944,
 *     "gifter":972,
 *     "length":2916,
 *     "area":"A_NG",
 *     "rating":"A",
 *     "exp":1458,
 *     "type":"video_rating"
 * }
 * 数据清洗作业
 * 需求：
 * 1：从原始数据（JSON格式）中过滤出来需要的字段
 *      主播ID(uid)、金币数量(gold)、总观看PV(watchnumpv)、粉丝关注数量(follwer)、视频总开播时长(length)
 * 2: 针对核心字段进行异常值判断
 *      金币数量、总观看PV、粉丝关注数量、视频总开播时长
 *      以上四个字段正常情况下都不应该是负值，页不应该缺失
 *      如果这些字段值为负值，则认为是异常数据，直接丢弃，如果这些字段值个别有缺失，则认为字段的值为0即可
 *
 * 分析：
 * 1：由于原始数据是json格式的，所以可以使用fastjson对原始数据进行解析，获取指定字段的内容
 * 2：如果对获取到的数据进行判断，只保留满足条件的数据即可
 * 3：由于不需要聚合过程，只是一个简单的过滤操作，所以只需要map阶段即可，reduce阶段就不需要了
 * 4：其中map阶段的k1,v1的数据类型是固定的：<LongWritable, Text>
 *     k2,v2的数据类型为：<Text, Text>， k2存储主播ID，v2存储核心字段多个字段中间用\t分割
 *
 *
 * @author guowm
 * @date 2022/7/6
 */
@Slf4j
public class DataCleanJob {

    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //指定我们使用的文件系统类型:
        conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
        FileSystem fileSystem = null;
        try {
            fileSystem = FileSystem.get(conf);
            run(args, fileSystem);
            fileSystem.close();
        } catch (IOException e) {
            e.printStackTrace();
            if (Objects.nonNull(fileSystem)) {
                try {
                    fileSystem.close();
                } catch (IOException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
    }

    public static void run(String[] args, FileSystem fileSystem) {
        // 可以指定job需要配置的参数
        final Configuration conf = fileSystem.getConf();
        try {
            final String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
            if (otherArgs.length < 2) {
                System.err.println("Usage: DataCleanJob <in> [<in>...] <out>");
                System.exit(2);
            }

            // 参数
            String[] inputDirs = Arrays.copyOf(otherArgs, otherArgs.length - 1);
            String outputDir = otherArgs[otherArgs.length - 1];

            // 创建一个job
            final Job job = Job.getInstance(conf);

            // 注意：这一行必须设置，否则在集群中执行的是找不到 DataCleanJob 这个类
            job.setJarByClass(DataCleanJob.class);

            // 指定输入路径(可以是文件，也可以是目录)
            for (int i = 0; i < inputDirs.length; i++) {
                FileInputFormat.addInputPath(job, new Path(inputDirs[i]));
            }
            // 指定输出路径(只能指定一个不存在的目录)
            FileOutputFormat.setOutputPath(job, new Path(outputDir));

            // 指定map相关的代码
            job.setMapperClass(DataCleanMap.class);
            // 指定k2的类型
            job.setMapOutputKeyClass(Text.class);
            // 指定v2的类型
            job.setMapOutputValueClass(Text.class);

            // 禁用reduce，因为用不到 reduce 聚合操作，所以可以禁用，节省资源
            job.setNumReduceTasks(0);

            // 提交job
            final boolean waitForCompletion = job.waitForCompletion(true);
            log.info("---execute result: {}---", waitForCompletion);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
